package examples.streaming;

import com.alibaba.fastjson.JSONObject;
import examples.entity.CsdnOrderEntity;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.net.URLDecoder;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * @Description
 * @Author wanghan
 * @Date 2020/10/15
 * @Version V1.0.0
 **/
public class CSDNOrderAnalyseMr {
    public static void main(String[] args) {
        if (args.length < 2) {
            System.err.println("Usage: CSDNOrderAnalyseMr <data_path>\n" +
                    "  <data_path> csdn order data file path\n\n");
            System.exit(1);
        }
        String filePath = args[0];
        String outFilePath = args[1];

        SparkSession sparkSession = SparkSession
                .builder()
                .appName("CSDNOrderAnalyseMr")
                .getOrCreate();

        JavaSparkContext jsc = new JavaSparkContext(sparkSession.sparkContext());
        JavaRDD<String> logRdd = jsc.textFile(filePath);
        JavaRDD<CsdnOrderEntity> orderEntityRDD = logRdd.map(s -> {
            try {
                return parseToOrder(s);
            } catch (Exception e) {
                return null;
            }
        }).filter(e -> Objects.nonNull(e));

        Dataset<Row> orderDs = sparkSession.createDataFrame(orderEntityRDD, CsdnOrderEntity.class);
        /**
         * root
         *  |-- createTime: long (nullable = false)
         *  |-- flag: integer (nullable = false)
         *  |-- goodsId: integer (nullable = false)
         *  |-- goodsName: string (nullable = true)
         *  |-- orderNo: string (nullable = true)
         *  |-- payType: string (nullable = true)
         *  |-- totalPrice: long (nullable = false)
         *  |-- username: string (nullable = true)
         */
        orderDs.printSchema();
        //将dataset注册为临时表
        orderDs.createOrReplaceTempView("order");
        //查看总记录数
        sparkSession.sql("select count(*) as c from order").show();

        //按照销量获取商品排行
        Dataset<Row> resultRdd01 = sparkSession.sql("select goodsName ,count(0) as sellCount  from order group by goodsName order by sellCount desc");

        //按照销售金额获取商品排行
        Dataset<Row> resultRdd02 = sparkSession.sql("select goodsName ,sum(totalPrice/100.0) as GMV  from order group by goodsName order by GMV desc");

        //获取不同支付方式的使用次数排行
        Dataset<Row> resultRdd03 = sparkSession.sql("select payType ,count(0) as payTypeCount  from order group by payType order by payTypeCount desc");

        resultRdd01.repartition(1).write().json(outFilePath + "/goods_sell_rank");
        resultRdd02.repartition(1).write().json(outFilePath + "/goods_gmv_rank");
        resultRdd03.repartition(1).write().json(outFilePath + "/paytype_rank");

    }

    private static CsdnOrderEntity parseToOrder(String log) throws Exception {
        //171.11.101.93 - [15/Oct/2020:15:00:01 +0800] "GET /?order_no=11602745189049h5TGquzz8&goods_name=VIP%E4%B8%A4%E5%B9%B4%E5%8D%A1&flag=4&create_time=1602745189049&total_price=49900&goods_id=10004&pay_type=WECHAT_PAY&username=weixin_50079947 HTTP/1.1" 200 12 "Java/1.8.0_231"
        if (StringUtils.isBlank(log)) {
            return null;
        }
        String logParams[] = log.split(" ");
        if (logParams.length < 10) {
            return null;
        }
        ///?order_no=11602745189049h5TGquzz8&goods_name=VIP%E4%B8%A4%E5%B9%B4%E5%8D%A1&flag=4&create_time=1602745189049&total_price=49900&goods_id=10004&pay_type=WECHAT_PAY&username=weixin_50079947
        String orderParam = URLDecoder.decode(logParams[5], "UTF-8").replaceAll("/\\?", "");
        Map<String, Object> orderParamMap = new HashMap<>();
        Arrays.stream(orderParam.split("&")).forEach(p -> {
            String paramKv[] = p.split("=");
            if (paramKv.length == 2) {
                orderParamMap.put(paramKv[0], paramKv[1]);
            }
        });
        return JSONObject.parseObject(JSONObject.toJSONString(orderParamMap), CsdnOrderEntity.class);
    }
}
